package com.yifeng.repo.controller.kafka.message.consume;

import com.talkyun.utils.revised.Looper;
import com.yifeng.repo.base.api.TimeRange;
import com.yifeng.repo.base.constant.BaseConstant;
import com.yifeng.repo.controller.kafka.message.handler.KafkaHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Collection;
import java.util.List;
import java.util.function.Consumer;

/**
 * @author 00115764
 * Created by daibing on 2024/5/2.
 */
@Slf4j
public class SubscribePullConsumerLooper extends Looper {
    private final KafkaConsumer<String, String> pullConsumer;
    private final List<TimeRange> timeRangeList;
    private final Consumer<ConsumerRecords<String, String>> msgListConsumer;
    private long lastPrintTime = 0L;
    private boolean paused = false;

    public SubscribePullConsumerLooper(String name, int interval, int errorInterval,
                                       KafkaConsumer<String, String> pullConsumer, List<TimeRange> timeRangeList,
                                       Consumer<ConsumerRecords<String, String>> msgListConsumer) {
        super(name, interval, errorInterval);
        this.pullConsumer = pullConsumer;
        this.timeRangeList = timeRangeList;
        this.msgListConsumer = msgListConsumer;
        log.info("SubscribePullConsumerLooper init: name={}, timeRangeList={}, partitions={}",
                name, timeRangeList, KafkaHelper.buildPartitionLog(pullConsumer.assignment()));
    }

    @Override
    protected void loop() throws Throwable {
        // 检查当前时间段
        boolean nowInside = TimeRange.nowInside(timeRangeList);
        if (nowInside) {
            // 1.如果已经暂停，就恢复消费，并且设置状态标记
            if (paused) {
                pullConsumer.resume(pullConsumer.paused());
                paused = false;
            }

            // 2.拉取&消费消息
            ConsumerRecords<String, String> consumerRecords = pullConsumer.poll(Duration.ofMillis(5 * 1000));
            msgListConsumer.accept(consumerRecords);

            // 3. 间歇性打印消费分区信息
            if (lastPrintTime + BaseConstant.ONE_HOUR_MS < System.currentTimeMillis()) {
                log.info("SubscribePullConsumerLooper ok: name={}, timeRangeList={}, assignment={}, size={}",
                        getName(), timeRangeList, KafkaHelper.buildPartitionLog(pullConsumer.assignment()), consumerRecords.count());
                lastPrintTime = System.currentTimeMillis();
            }
        } else {
            // 1.如果已经暂停，继续睡眠1分钟
            if (paused) {
                doSleep(60 * 1000);
                return;
            }
            // 2.如果没有暂停，就暂停，并且设置状态标记，以及睡眠1分钟
            pullConsumer.pause(pullConsumer.assignment());
            paused = true;
            doSleep(60 * 1000);
        }
    }

    @Override
    protected void loopThrowable(Throwable t) {
        log.warn("consumer {} failed: error=", getName(), t);
    }

    @Override
    protected void clear() {
        pullConsumer.unsubscribe();
        pullConsumer.close();
    }

    public KafkaConsumer<String, String> getPullConsumer() {
        return pullConsumer;
    }

}
